package com.beyondprototype.batch.config;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.batch.item.support.SingleItemPeekableItemReader;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import com.beyondprototype.batch.model.CustomerMarker;
import com.beyondprototype.batch.model.PremierCustomer;
import com.beyondprototype.batch.service.PremierMarkerService;

import lombok.extern.slf4j.Slf4j;

@Configuration
@Slf4j
public class GcduFlatDataPeekableBatchConfiguration {

	
	@Bean
    @Qualifier("gcduFlatPeekableItemReader")
    public ItemReader<CustomerMarker> gcduFlatPeekableItemReader(@Qualifier("gcduFlatItemReader") ItemReader<CustomerMarker> delegate) {
    	
		 SingleItemPeekableItemReader<CustomerMarker> reader = new SingleItemPeekableItemReader<>();
		 
		 ExecutionContext executionContext = new ExecutionContext();
		 
		 reader.setDelegate(delegate);
		 
		 reader.open(executionContext);
		
		
		 return new ItemReader<CustomerMarker> () {

			int count = 0;
			
			@Override
			public CustomerMarker read()
					throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
				
				CustomerMarker marker = reader.read();
				
				if(marker != null) {//not end of the file, no more record to process
					
					count++;
					
					CustomerMarker nextMarker = reader.peek();
					
					if(nextMarker == null) {//current record is the last record in the file
						marker.setStatus("Last");
					}
					else if(!marker.getGlobalIdentifier().equals(nextMarker.getGlobalIdentifier())) {
						//current record is the last record for the same customer
						marker.setStatus("Last");
					}
					
					log.info(String.format("current : %s, next: %s", marker.getGlobalIdentifier(),
							nextMarker == null ? marker.getGlobalIdentifier() : nextMarker.getGlobalIdentifier()));
				}
				
				log.info(String.format("Number of records : %s", count));
				
				return marker;
			}
		 };
    }
	 
	
    
    @Bean
    @Qualifier("gcduFlat2JsonItemProcessor")
    public ItemProcessor<CustomerMarker, PremierCustomer> gcduFlat2JsonItemProcessor(@Qualifier("globalCustomerCache") Map<String, PremierCustomer> cache) {
    	
        return new ItemProcessor<CustomerMarker, PremierCustomer>() {
        	        	
            @Override
            public PremierCustomer process(CustomerMarker item) throws Exception {
            	
            	log.info("Processing "+item.toString());
            	
            	boolean isLast = "Last".equals(item.getStatus());
            	
            	log.info("isLast = " + isLast +" ," + item.getGlobalIdentifier());
            	
            	PremierCustomer premierMarker = cache.get(item.getGlobalIdentifier());
            	
            	if(premierMarker == null) {
            		//initialize the current global customer
            		premierMarker = new PremierCustomer();
            		premierMarker.setGlobalIdentifier(item.getGlobalIdentifier());
            		premierMarker.setMarkers(new ArrayList<>());
            		premierMarker.getMarkers().add(item);
            		
            		cache.put(premierMarker.getGlobalIdentifier(), premierMarker);
            		
            	}else if(premierMarker.getGlobalIdentifier().equals(item.getGlobalIdentifier())) {
            		//add to current global customer
            		premierMarker.getMarkers().add(item);
            	}
            	
            	if(isLast) {
            		
            		//TODO: calculate P1PA for the customer 
            		
            		premierMarker.setStatus("Done");
            		
            		cache.remove(premierMarker.getGlobalIdentifier());
            		
            		return premierMarker;
            	}
            	
            	return null;
            }
        };
    }
    
    @Bean 
    @Qualifier("gcduFlat2JsonItemWriter")
    public ItemWriter<PremierCustomer> gcduFlat2JsonItemWriter(PremierMarkerService service) {
    	
    	PremierMarkerService premierMarkerService = service;
    	
        return new ItemWriter<PremierCustomer>() {
            @Override
            public void write(List<? extends PremierCustomer> items) throws Exception {
            	
            	List<PremierCustomer> doneItems = new ArrayList<>();
            	
                for (PremierCustomer item : items) {
                	
                	if(item != null) {
                		doneItems.add(item);
                	}
                }
                
                if(!doneItems.isEmpty()) {
                	premierMarkerService.create(doneItems);
                }
            }
        };
    }
    
    
    @Bean
    @Qualifier("globalCustomerCache")
    public Map<String, PremierCustomer> cache() {
    	return new ConcurrentHashMap<String, PremierCustomer>(2);
    }
}
